from sqlalchemy.orm.session import Session

from airflow.exceptions import AirflowException
from airflow.models.idata_folder import Folder
from airflow.models.idata_task import Task
from airflow.models.idata_workflow import WorkFlow
from airflow.utils.session import provide_session


@provide_session
def delete(folder_id, session: Session = None):
    sub = Folder.query_children(folder_id=folder_id, session=session)
    tasks = Task.query_task(folder_id=folder_id, session=session)
    wfs = WorkFlow.query_workflow(folder_id=folder_id, session=session)
    if sub or tasks or wfs:
        raise AirflowException('folder is not empty')
    session.query(Folder).filter(Folder.id == folder_id).delete()
